// writers/kwdb_writer.go
package writers

import (
	"bytes"
	"database/sql"
	"encoding/hex"
	"fmt"
	"log"
	"mig2kwdb/common"
	"net"
	"reflect"
	"strconv"
	"strings"

	pq "github.com/lib/pq"
)

type KwdbWriter struct {
	db *sql.DB
}

func (w *KwdbWriter) GetDB() *sql.DB {
	return w.db
}

func (w *KwdbWriter) Close() {
	if w.db != nil {
		w.db.Close()
	}
}

func BufferQuoteIdentifierString(name string) string {
	buffer := bytes.NewBufferString("")
	end := strings.IndexRune(name, 0)
	if end > -1 {
		name = name[:end]
	}
	buffer.WriteRune('"')
	buffer.WriteString(strings.Replace(name, `"`, `""`, -1))
	buffer.WriteRune('"')
	return buffer.String()
}
func BufferQuoteIdentifier(name string, buffer *bytes.Buffer) {
	end := strings.IndexRune(name, 0)
	if end > -1 {
		name = name[:end]
	}
	buffer.WriteRune('"')
	buffer.WriteString(strings.Replace(name, `"`, `""`, -1))
	buffer.WriteRune('"')
}
func makeStmt(buffer *bytes.Buffer, columns ...string) {
	//s := bytes.NewBufferString()
	for i, col := range columns {
		if i != 0 {
			buffer.WriteString(", ")
		}
		BufferQuoteIdentifier(col, buffer)
	}
	buffer.WriteString(") FROM STDIN")
}
func getFullTableName(dbName, schemaName, tableName string) string {
	buffer := bytes.NewBufferString("")
	BufferQuoteIdentifier(dbName, buffer)
	buffer.WriteString(".")
	BufferQuoteIdentifier(schemaName, buffer)
	buffer.WriteString(".")
	BufferQuoteIdentifier(tableName, buffer)
	return buffer.String()
}

// KwdbWriter 创建新的 KwdbWriter 实例并初始化数据库连接
func NewKwdbWriter(connInfo common.DBConnectionInfo) (*KwdbWriter, error) {
	dsn := "postgres://" + connInfo.User + ":" + connInfo.Password + "@" + connInfo.Ip + ":" + fmt.Sprint(connInfo.Port) + "/" + connInfo.DatabaseName + "?"

	if connInfo.SSL {
		dsn += "sslmode=require"
	} else {
		dsn += "sslmode=disable"
	}
	conn, err := sql.Open("postgres", dsn)
	//config, err := pgx.ParseConfig(dsn)
	//db := stdlib.OpenDB(*config)

	return &KwdbWriter{
		db: conn,
	}, err
}
func (w *KwdbWriter) IsNetworkError(err error) bool {
	log.Println(err)
	if _, ok := err.(net.Error); ok {
		return true
	}
	etype := reflect.TypeOf(err)
	name := etype.String()
	if strings.EqualFold(name, "*fmt.wrapError") || strings.EqualFold(name, "pgx.PgError") ||
		strings.EqualFold(name, "*pq.Error") || strings.EqualFold(name, "*errors.errorString") {
		return false
	} else {
		return true
	}
}
func (w *KwdbWriter) ExistSDatabase(dbName string) bool {
	// 构建查询数据库是否存在的SQL语句
	var sql string

	sql = fmt.Sprintf("SELECT 1 FROM pg_database WHERE datname = '%s'", dbName)

	// 执行查询
	rows, err := w.db.Query(sql)
	if err != nil {
		log.Printf("Error checking if database %s exists: %v", dbName, err)
		return false
	}
	defer rows.Close()

	// 如果查询结果有行，则数据库存在
	return rows.Next()
}
func (w *KwdbWriter) CreateDatabase(dbName string, IsSeries bool) error {
	if w.ExistSDatabase(dbName) {
		return nil
	}
	buffer := bytes.NewBufferString("CREATE DATABASE IF NOT EXISTS ")
	if IsSeries {
		buffer = bytes.NewBufferString("CREATE TS DATABASE ")
	}
	BufferQuoteIdentifier(dbName, buffer)
	_, err := w.db.Exec(buffer.String())
	if err != nil {
		if pqErr, ok := err.(*pq.Error); ok && pqErr.Code == "42P04" {
			log.Println("Error creating database:", err)
			return nil
		}
		log.Println("Error creating database:", err)
		return err
	}
	return nil
}
func (w *KwdbWriter) CreateSchema(dbName, schemaName string, IsSeries bool) error {
	if IsSeries {
		log.Println("skip creating schema")
		return nil
	}
	buffer := bytes.NewBufferString("use ")
	BufferQuoteIdentifier(dbName, buffer)
	buffer.WriteString("; CREATE SCHEMA IF NOT EXISTS ")
	BufferQuoteIdentifier(schemaName, buffer)
	_, err := w.db.Exec(buffer.String())
	if err != nil {
		log.Println("Error creating schema:", err)
		return err
	}

	return nil
}
func (w *KwdbWriter) CreateTable(dbName, schemaName, tableName string, tableDef common.TableDefinition, IsSeries bool) error {
	// build sql
	fullTableName := getFullTableName(dbName, schemaName, tableName)
	createSQL := w.buildCreateTableSQL(fullTableName, tableDef, IsSeries)

	// 执行创建表语句
	_, err := w.db.Exec(createSQL)
	if err != nil {
		return fmt.Errorf("failed to create table %s: %v,%s", tableName, err, createSQL)
	}
	return nil
}

func (w *KwdbWriter) ConvertBinaryArrayString(input string) string {
	// 去除前后的花括号
	trimmed := strings.Trim(input, "{}")

	// 如果是空数组，直接返回空数组格式
	if trimmed == "" {
		return "ARRAY[]"
	}

	// 按逗号分割二进制值
	binaryValues := strings.Split(trimmed, ",")

	// 构建结果字符串
	var result strings.Builder
	result.WriteString("ARRAY[")

	for i, val := range binaryValues {
		if i > 0 {
			result.WriteString(",")
		}
		result.WriteString(fmt.Sprintf("B'%s'", val))
	}

	result.WriteString("]")
	return result.String()
}
func (w *KwdbWriter) GetFieldArrayValue(dataType string, sqlValue *sql.NullString) interface{} {
	switch strings.ToUpper(dataType) {
	case "BIT VARYING[]", "BIT(1)[]":
		return w.ConvertBinaryArrayString(sqlValue.String)

	}
	return "'" + sqlValue.String + "'"
}
func (w *KwdbWriter) GetFieldValue(dataType string, sqlValue *sql.NullString) interface{} {

	strType := strings.ToUpper(dataType)
	if strings.Contains(strType, "[]") {
		return w.GetFieldArrayValue(strType, sqlValue)
	}
	switch strType {
	case "BYTEA":
		if sqlValue.Valid && !strings.HasPrefix(sqlValue.String, "\\x") {
			return []byte(sqlValue.String)
		} else {
			hexStr := sqlValue.String
			if strings.HasPrefix(hexStr, `\x`) {
				hexStr = hexStr[2:]
			} else if strings.HasPrefix(hexStr, `0x`) {
				hexStr = hexStr[2:]
			}
			res, _ := hex.DecodeString(hexStr)
			return res
		}
	case "BIT":
		if sqlValue.Valid {
			if strings.EqualFold("true", strings.ToLower(sqlValue.String)) {
				return "1"
			}
			return "0"
		} else {
			return "0"
		}
	case "INTEGER", "INT", "REAL", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT", "DECIMAL", "DOUBLE", "FLOAT":
		if sqlValue.Valid {
			return sqlValue.String
		} else {
			return nil
		}
	case "GEOGRAPHY", "GEOMETRY":
		if sqlValue.Valid {
			hstr := hex.EncodeToString([]byte(sqlValue.String))
			return "0x" + strings.ToUpper(hstr)
		}
		return nil
	case "BINARY", "HIERARCHYID", "IMAGE", "VARBINARY", "UNIQUEIDENTIFIER":
		if sqlValue.Valid {
			hstr := hex.EncodeToString([]byte(sqlValue.String))
			return "0x" + hstr
		}
		return ""
	case "BLOB":
		return sqlValue.String
	default:
		if sqlValue.Valid {
			return sqlValue.String
		} else {
			return nil
		}
	}
}
func (w *KwdbWriter) GetFieldValueFromReplicate(dataType string, sqlValue interface{}) interface{} {
	strType := strings.ToUpper(dataType)
	switch strType {
	case "DECIMAL", "DOUBLE", "FLOAT":
		if sqlValue != nil {
			return fmt.Sprint(sqlValue)
		}
		return "NULL"
	}

	return sqlValue

}

// DeleteRecord 根据主键删除单条记录
func (w *KwdbWriter) DeleteRecord(dbName, schemaName, tableName string, columns []common.KwdbColumn, value []interface{}, from string, IsSeries bool) (uint64, error) {
	// 构建完整的表名
	fullTableName := getFullTableName(dbName, schemaName, tableName)

	// 构建DELETE语句
	buffer := bytes.NewBufferString("DELETE FROM ")
	buffer.WriteString(fullTableName)

	// 查找主键列
	var primaryKeyColumns []common.KwdbColumn
	var primaryKeyValues []interface{}
	var args []interface{}
	argIndex := 1

	// 遍历所有列，找到主键列
	for i, col := range columns {
		if col.PrimaryKey.String == "YES" {
			primaryKeyColumns = append(primaryKeyColumns, col)
			primaryKeyValues = append(primaryKeyValues, value[i])
		}
	}

	// 检查是否存在主键
	if len(primaryKeyColumns) == 0 {
		return 0, fmt.Errorf("table %s has no primary key, cannot perform delete operation", tableName)
	}

	// 构建WHERE子句
	buffer.WriteString(" WHERE ")
	for i, col := range primaryKeyColumns {
		if i > 0 {
			buffer.WriteString(" AND ")
		}
		BufferQuoteIdentifier(col.Name, buffer)
		buffer.WriteString(fmt.Sprintf(" = $%d", argIndex))

		// 处理主键值
		if from == "sync" {
			fieldValue := w.GetFieldValue(col.Type, primaryKeyValues[i].(*sql.NullString))
			args = append(args, fieldValue)
		} else {
			fieldValue := w.GetFieldValueFromReplicate(col.Type, primaryKeyValues[i])
			args = append(args, fieldValue)
		}
		argIndex++
	}

	// 开始事务
	transaction, err := w.db.Begin()
	if err != nil {
		return 0, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() {
		if err != nil {
			transaction.Rollback()
		}
	}()

	// 执行删除语句
	result, err := transaction.Exec(buffer.String(), args...)
	if err != nil {
		return 0, fmt.Errorf("failed to execute delete statement: %w, SQL: %s", err, buffer.String())
	}

	// 提交事务
	if err = transaction.Commit(); err != nil {
		return 0, fmt.Errorf("failed to commit transaction: %w", err)
	}

	// 获取受影响的行数
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		// 某些数据库驱动可能不支持RowsAffected
		log.Printf("Warning: failed to get rows affected: %v", err)
		return 0, nil
	}

	return uint64(rowsAffected), nil
}

// UpdateReplicateRecords 更新复制记录，每两个记录为一对
// 第一条记录为原始记录（用于WHERE条件），第二条记录为更新后的记录（用于SET所有字段的值）
func (w *KwdbWriter) UpdateReplicateRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}, from string) (uint64, error) {
	// 检查记录数是否为偶数
	if len(values)%2 != 0 {
		return 0, fmt.Errorf("values count must be even, got %d", len(values))
	}

	var totalRowsAffected uint64 = 0

	// 开始事务
	transaction, err := w.db.Begin()
	if err != nil {
		return 0, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() {
		if err != nil {
			transaction.Rollback()
		}
	}()

	// 处理每一对记录
	for i := 0; i < len(values); i += 2 {
		originalRecord := values[i]  // 原始记录（用于WHERE条件）
		updatedRecord := values[i+1] // 更新后的记录（用于SET所有字段的值）

		// 构建完整的表名
		fullTableName := getFullTableName(dbName, schemaName, tableName)

		// 构建UPDATE语句
		buffer := bytes.NewBufferString("UPDATE ")
		buffer.WriteString(fullTableName)
		buffer.WriteString(" SET ")

		// 构建SET子句（仅包含有变化的字段）
		var args []interface{}
		argIndex := 1
		changedColumns := make([]int, 0) // 记录有变化的字段索引

		// 比较每个字段，只添加有变化的字段到更新列表
		for j, col := range columns {
			var originalValue, updatedValue interface{}

			// 获取原始记录和更新后记录的值
			if from == "sync" {
				originalSqlValue, ok := originalRecord[j].(*sql.NullString)
				if !ok {
					return 0, fmt.Errorf("cannot convert original column %s value to sql.NullString", col.Name)
				}
				updatedSqlValue, ok := updatedRecord[j].(*sql.NullString)
				if !ok {
					return 0, fmt.Errorf("cannot convert updated column %s value to sql.NullString", col.Name)
				}

				// 比较两个值是否相同
				if originalSqlValue.Valid != updatedSqlValue.Valid ||
					(originalSqlValue.Valid && updatedSqlValue.Valid && originalSqlValue.String != updatedSqlValue.String) {
					changedColumns = append(changedColumns, j)
				}

				originalValue = originalSqlValue
				updatedValue = updatedSqlValue
			} else {
				originalValue = originalRecord[j]
				updatedValue = updatedRecord[j]

				// 比较两个值是否相同（简单比较，实际可能需要更复杂的逻辑）
				if fmt.Sprint(originalValue) != fmt.Sprint(updatedValue) {
					changedColumns = append(changedColumns, j)
				}
			}
		}

		// 如果没有字段变化，跳过此次更新
		if len(changedColumns) == 0 {
			continue
		}

		// 添加有变化的字段到SET子句
		for idx, j := range changedColumns {
			col := columns[j]
			if idx > 0 {
				buffer.WriteString(", ")
			}
			BufferQuoteIdentifier(col.Name, buffer)
			buffer.WriteString(fmt.Sprintf(" = $%d", argIndex))

			// 处理字段值（使用更新后记录的值）
			if from == "sync" {
				sqlValue, ok := updatedRecord[j].(*sql.NullString)
				if !ok {
					return 0, fmt.Errorf("cannot convert column %s value to sql.NullString", col.Name)
				}
				fieldValue := w.GetFieldValue(col.Type, sqlValue)
				args = append(args, fieldValue)
			} else {
				fieldValue := w.GetFieldValueFromReplicate(col.Type, updatedRecord[j])
				args = append(args, fieldValue)
			}
			argIndex++
		}

		// 构建WHERE子句（使用原始记录的主键值定位记录）
		buffer.WriteString(" WHERE ")
		firstPK := true
		for j, col := range columns {
			if col.PrimaryKey.String == "YES" {
				if !firstPK {
					buffer.WriteString(" AND ")
				}
				BufferQuoteIdentifier(col.Name, buffer)
				buffer.WriteString(fmt.Sprintf(" = $%d", argIndex))

				// 处理主键值（使用原始记录的主键值）
				if from == "sync" {
					sqlValue, ok := originalRecord[j].(*sql.NullString)
					if !ok {
						return 0, fmt.Errorf("cannot convert column %s value to sql.NullString", col.Name)
					}
					fieldValue := w.GetFieldValue(col.Type, sqlValue)
					args = append(args, fieldValue)
				} else {
					fieldValue := w.GetFieldValueFromReplicate(col.Type, originalRecord[j])
					args = append(args, fieldValue)
				}
				argIndex++
				firstPK = false
			}
		}

		// 检查是否有主键
		if firstPK {
			return 0, fmt.Errorf("table %s has no primary key, cannot perform update operation", tableName)
		}

		// 执行更新语句
		result, err := transaction.Exec(buffer.String(), args...)
		if err != nil {
			return 0, fmt.Errorf("failed to execute update statement: %w, SQL: %s", err, buffer.String())
		}

		// 获取受影响的行数
		rowsAffected, err := result.RowsAffected()
		if err != nil {
			log.Printf("Warning: failed to get rows affected: %v", err)
		} else {
			totalRowsAffected += uint64(rowsAffected)
		}
	}

	// 提交事务
	if err = transaction.Commit(); err != nil {
		return 0, fmt.Errorf("failed to commit transaction: %w", err)
	}

	return totalRowsAffected, nil
}

// UpdateRecord 根据主键更新单条记录
func (w *KwdbWriter) UpdateRecord(dbName, schemaName, tableName string, columns []common.KwdbColumn, value []interface{}, from string, IsSeries bool) (uint64, error) {
	// 构建完整的表名
	fullTableName := getFullTableName(dbName, schemaName, tableName)

	// 构建UPDATE语句
	buffer := bytes.NewBufferString("UPDATE ")
	buffer.WriteString(fullTableName)
	buffer.WriteString(" SET ")

	// 分离主键列和非主键列
	var primaryKeyColumns []common.KwdbColumn
	var primaryKeyValues []interface{}
	var nonPrimaryKeyColumns []common.KwdbColumn
	var nonPrimaryKeyValues []interface{}
	var args []interface{}
	argIndex := 1

	// 遍历所有列，分离主键和非主键
	for i, col := range columns {
		// 确保值的类型正确
		sqlValue, ok := value[i].(*sql.NullString)
		if !ok {
			log.Printf("Warning: cannot convert column %s value to sql.NullString", col.Name)
			continue
		}

		if col.PrimaryKey.String == "YES" {
			primaryKeyColumns = append(primaryKeyColumns, col)
			primaryKeyValues = append(primaryKeyValues, sqlValue)
		} else {
			nonPrimaryKeyColumns = append(nonPrimaryKeyColumns, col)
			nonPrimaryKeyValues = append(nonPrimaryKeyValues, sqlValue)
		}
	}

	// 检查是否存在主键
	if len(primaryKeyColumns) == 0 {
		return 0, fmt.Errorf("table %s has no primary key, cannot perform update operation", tableName)
	}

	// 检查是否存在需要更新的非主键列
	if len(nonPrimaryKeyColumns) == 0 {
		// 没有需要更新的列，直接返回
		return 0, nil
	}

	// 构建SET子句
	for i, col := range nonPrimaryKeyColumns {
		if i > 0 {
			buffer.WriteString(", ")
		}
		BufferQuoteIdentifier(col.Name, buffer)
		buffer.WriteString(fmt.Sprintf(" = $%d", argIndex))

		// 处理字段值
		if from == "sync" {
			fieldValue := w.GetFieldValue(col.Type, nonPrimaryKeyValues[i].(*sql.NullString))
			args = append(args, fieldValue)
		} else {
			fieldValue := w.GetFieldValueFromReplicate(col.Type, nonPrimaryKeyValues[i])
			args = append(args, fieldValue)
		}
		argIndex++
	}

	// 构建WHERE子句
	buffer.WriteString(" WHERE ")
	for i, col := range primaryKeyColumns {
		if i > 0 {
			buffer.WriteString(" AND ")
		}
		BufferQuoteIdentifier(col.Name, buffer)
		buffer.WriteString(fmt.Sprintf(" = $%d", argIndex))

		// 处理主键值
		if from == "sync" {
			fieldValue := w.GetFieldValue(col.Type, primaryKeyValues[i].(*sql.NullString))
			args = append(args, fieldValue)
		} else {
			fieldValue := w.GetFieldValueFromReplicate(col.Type, primaryKeyValues[i])
			args = append(args, fieldValue)
		}
		argIndex++
	}

	// 开始事务
	transaction, err := w.db.Begin()
	if err != nil {
		return 0, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() {
		if err != nil {
			transaction.Rollback()
		}
	}()

	// 执行更新语句
	result, err := transaction.Exec(buffer.String(), args...)
	if err != nil {
		return 0, fmt.Errorf("failed to execute update statement: %w, SQL: %s", err, buffer.String())
	}

	// 提交事务
	if err = transaction.Commit(); err != nil {
		return 0, fmt.Errorf("failed to commit transaction: %w", err)
	}

	// 获取受影响的行数
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		log.Printf("Warning: failed to get rows affected: %v", err)
		return 0, nil
	}

	return uint64(rowsAffected), nil
}
func (w *KwdbWriter) InsertSeriesRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}) (uint64, error) {
	var totalDataSize uint64 = 0
	var columnNames []string

	// 获取列名
	for _, column := range columns {
		columnNames = append(columnNames, column.Name)
	}

	// 开始构建INSERT语句
	fullTableName := getFullTableName(dbName, schemaName, tableName)
	buffer := bytes.NewBufferString("INSERT INTO ")
	buffer.WriteString(fullTableName)
	buffer.WriteString(" (")

	// 添加列名
	for i, columnName := range columnNames {
		if i > 0 {
			buffer.WriteString(", ")
		}
		BufferQuoteIdentifier(columnName, buffer)
	}
	buffer.WriteString(") VALUES ")

	// 构建值部分
	valuePlaceholders := make([]string, len(values))
	args := make([]interface{}, 0)
	argIndex := 1

	for i, row := range values {
		rowBuffer := bytes.NewBufferString("(")
		for j, v := range row {
			if j > 0 {
				rowBuffer.WriteString(", ")
			}

			sqlValue, ok := v.(*sql.NullString)
			if !ok {
				log.Println("cannot convert to sql.NullString")
				continue
			}

			fieldValue := w.GetFieldValue(columns[j].Type, sqlValue)
			if sqlValue.Valid {
				totalDataSize += uint64(len(sqlValue.String))
			}

			// 使用占位符来避免SQL注入
			rowBuffer.WriteString(fmt.Sprintf("$%d", argIndex))
			args = append(args, fieldValue)
			argIndex++
		}
		rowBuffer.WriteString(")")
		valuePlaceholders[i] = rowBuffer.String()
	}

	// 将所有值组合在一起
	buffer.WriteString(strings.Join(valuePlaceholders, ", "))

	// 执行批量插入
	transaction, err := w.db.Begin()
	if err != nil {
		log.Println("failed to begin transaction:", err)
		return 0, err
	}

	_, err = transaction.Exec(buffer.String(), args...)
	if err != nil {
		transaction.Rollback()
		log.Println("failed to execute batch insert:", err)
		return 0, err
	}

	err = transaction.Commit()
	if err != nil {
		log.Println("failed to commit transaction:", err)
		return 0, err
	}

	return totalDataSize, nil
}
func (w *KwdbWriter) InsertRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}, from string, IsSeries bool) (uint64, error) {
	if IsSeries {
		return w.InsertSeriesRecords(dbName, schemaName, tableName, columns, values)
	}
	var total_data_size uint64 = 0
	var ColumnNames []string
	for _, column := range columns {
		ColumnNames = append(ColumnNames, column.Name)
	}
	transaction, err := w.db.Begin()
	if err != nil {
		log.Println(err)
		return 0, err
	}

	fullTableName := getFullTableName(dbName, schemaName, tableName)
	buffer := bytes.NewBufferString("COPY " + fullTableName + " ")
	buffer.WriteString(" (")
	makeStmt(buffer, ColumnNames...)

	stmt, err := transaction.Prepare(buffer.String())
	if err != nil {
		log.Println(err)
		return 0, err
	}

	defer stmt.Close()
	for _, value := range values {
		var copyValues []interface{}
		for k, v := range value {

			if from == "sync" {
				sqlValue, ok := v.(*sql.NullString)
				if !ok {
					log.Println("connect convert sqlvalue")
					continue
				}

				fieldValue := w.GetFieldValue(columns[k].Type, sqlValue)
				if sqlValue.Valid {
					total_data_size += uint64(len(sqlValue.String))
				}
				copyValues = append(copyValues, fieldValue)

			} else {
				fieldValue := w.GetFieldValueFromReplicate(columns[k].Type, v)
				copyValues = append(copyValues, fieldValue)
			}

		}
		_, err = stmt.Exec(copyValues...)
		if err != nil {
			transaction.Rollback()
			return 0, err
		}

	}

	if _, err = stmt.Exec(); err != nil {
		//log.Println(err)
		transaction.Rollback()
		return 0, err
	}

	if err = transaction.Commit(); err != nil {
		//log.Println(err)
		return 0, nil
	}
	return total_data_size, err
}
func (w *KwdbWriter) InsertUpdateRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}, IsSeries bool) (uint64, error) {
	var oneSize uint64 = 0
	var err error
	var size uint64
	size, err = w.InsertRecords(dbName, schemaName, tableName, columns, values, "sync", IsSeries)
	if err != nil && strings.Contains(err.Error(), "duplicate key") {
		for _, value := range values {
			oneSize, err = w.InsertRecords(dbName, schemaName, tableName, columns, [][]interface{}{value}, "sync", IsSeries)
			if err != nil {
				if strings.Contains(err.Error(), "duplicate key") {
					oneSize, err = w.UpdateRecord(dbName, schemaName, tableName, columns, value, "sync", IsSeries)
					if err != nil {
						return size, err
					}
				}
			}
			size += oneSize
		}
	}
	return size, err
}
func (w *KwdbWriter) GetObjectName(name string) string {
	escaped := strings.ReplaceAll(name, `"`, `""`)
	return `"` + escaped + `"`
}
func (w *KwdbWriter) GetConditionRows(databaseName, schemaName, tableName string, columns []common.KwdbColumn, condition string) (common.RowsInterface, error) {
	orderby := ""
	keyNum := 0
	sql := "select "
	for idx, col := range columns {
		if idx > 0 {
			sql += ","
		}
		sql += " " + w.GetObjectName(col.Name)
		if col.PrimaryKey.String == "YES" {
			if keyNum > 0 {
				orderby += ","
			}
			orderby += " " + w.GetObjectName(col.Name)
			keyNum++
		}
	}
	// 如果没有主键，则使用唯一索引
	if orderby == "" {
		keyNum = 0
		for _, col := range columns {
			if col.Unique.String == "YES" {
				if keyNum > 0 {
					orderby += ","
				}
				orderby += " " + w.GetObjectName(col.Name)
				keyNum++
			}
		}
	}
	sql += " from " + w.GetObjectName(databaseName) + "." + w.GetObjectName(schemaName) + "." + w.GetObjectName(tableName)
	if condition != "" {
		sql += " where " + condition
	}
	if orderby != "" {
		sql += " order by " + orderby
	}

	rows, err := w.db.Query(sql)
	return common.WrapSQLRows(rows), err
}

func (w *KwdbWriter) DropTable(dbName, schemaName, tableName string, IsSeries bool) error {
	fullTableName := getFullTableName(dbName, schemaName, tableName)
	_, err := w.db.Exec("DROP TABLE IF EXISTS " + fullTableName + ";")
	return err
}

// buildCreateTableSQL 构建创建表的SQL语句，遵循开物数据库语法
func (w *KwdbWriter) buildCreateTableSQL(tableName string, tableDef common.TableDefinition, IsSeries bool) string {

	var tag_columns []common.KwdbColumn
	var sqlBuilder strings.Builder

	if IsSeries {
		sqlBuilder.WriteString(fmt.Sprintf("CREATE TABLE %s (\n", tableName))
	} else {
		sqlBuilder.WriteString(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (\n", tableName))
	}

	notFirst := false
	// 添加字段定义
	for _, col := range tableDef.Columns {
		if IsSeries && strings.ToUpper(col.Tag) == "YES" {
			tag_columns = append(tag_columns, col)
			continue
		}
		if notFirst {
			sqlBuilder.WriteString(",\n")
		}

		sqlBuilder.WriteString("  ")
		sqlBuilder.WriteString(BufferQuoteIdentifierString(col.Name))
		sqlBuilder.WriteString(" ")
		sqlBuilder.WriteString(w.mapDataType(col))

		// 处理非空约束
		if col.Nullable.Valid && col.Nullable.String == "NO" {
			sqlBuilder.WriteString(" NOT NULL")
		}
		if IsSeries && (strings.ToUpper(col.Type) == "TIMESTAMP" || strings.ToUpper(col.Type) == "TIMESTAMPTZ") {
			sqlBuilder.WriteString(" NOT NULL")
		}

		// 处理默认值
		if col.DefaultValue.Valid && col.DefaultValue.String != "" {
			sqlBuilder.WriteString(" DEFAULT ")
			sqlBuilder.WriteString(w.formatDefaultValue(col.DefaultValue.String, col.Type))
		}
		notFirst = true
	}
	if IsSeries && len(tag_columns) > 0 {
		//TAGs
		sqlBuilder.WriteString(") TAGS (")
		for i, col := range tag_columns {
			if i > 0 {
				sqlBuilder.WriteString(",")
			}
			sqlBuilder.WriteString("\"" + col.Name + "\"")
			sqlBuilder.WriteString(w.mapDataType(col))
			sqlBuilder.WriteString(" NOT NULL")
		}

		sqlBuilder.WriteString(")")
		sqlBuilder.WriteString("PRIMARY TAGS (")
		for i, col := range tag_columns {
			if i > 0 {
				sqlBuilder.WriteString(",")
			}
			sqlBuilder.WriteString("\"" + col.Name + "\"")
		}
		sqlBuilder.WriteString(")")
		//keep duration
		if tableDef.Properties["keep"] != nil {
			sqlBuilder.WriteString("retentions ")
			sqlBuilder.WriteString(tableDef.Properties["keep"].(string))
		}
		return sqlBuilder.String()
	}
	// 处理主键
	pkColumns := w.getPrimaryKeyColumns(tableDef.Columns)
	if len(pkColumns) > 0 {
		sqlBuilder.WriteString(",\n  PRIMARY KEY (")
		sqlBuilder.WriteString(strings.Join(pkColumns, ", "))
		sqlBuilder.WriteString(")")
	}

	sqlBuilder.WriteString(")")

	return sqlBuilder.String()
}
func (w *KwdbWriter) ReplicateRecords(dbName, schemaName, tableName string, columns []common.KwdbColumn, values [][]interface{}, sourceType, replicateType string) (uint64, error) {
	var oneSize uint64 = 0
	var err error
	var size uint64
	switch replicateType {
	case "insert":
		size, err = w.InsertRecords(dbName, schemaName, tableName, columns, values, sourceType, false)
		if err != nil && strings.Contains(err.Error(), "duplicate key") {
			for _, value := range values {
				oneSize, err = w.InsertRecords(dbName, schemaName, tableName, columns, [][]interface{}{value}, sourceType, false)
				if err != nil {
					if strings.Contains(err.Error(), "duplicate key") {
						oneSize, err = w.UpdateRecord(dbName, schemaName, tableName, columns, value, sourceType, false)
						if err != nil {
							return size, err
						}
					}
				}
				size += oneSize
			}
		}
	case "update":
		_, err := w.UpdateReplicateRecords(dbName, schemaName, tableName, columns, values, sourceType)
		if err != nil {
			log.Println(err)
			return size, err
		}
	case "delete":
		for _, value := range values {
			_, err := w.DeleteRecord(dbName, schemaName, tableName, columns, value, sourceType, false)
			if err != nil {
				log.Println(err)
				return size, err
			}
		}

	}
	return size, err
}
func (w *KwdbWriter) GetDataBaseType() string {
	return "kwdb"
}

// mapDataType 将MySQL数据类型映射到开物数据库类型
func (w *KwdbWriter) mapDataType(col common.KwdbColumn) string {

	switch strings.ToUpper(col.Type) {
	case "DECIMAL":
		if col.Precision > 0 || (col.Scale.Valid && col.Scale.String != "0") {
			if col.Scale.Valid {
				return fmt.Sprintf("DECIMAL(%d, %s)", col.Precision, col.Scale.String)
			}
			return fmt.Sprintf("DECIMAL(%d, 0)", col.Precision)
		}
		return "DECIMAL"
	case "CHAR", "VARCHAR", "CHARACTER", "NCHAR", "NVARCHAR":
		if col.Length.Valid {
			length, _ := strconv.ParseInt(col.Length.String, 10, 64)

			if length > 0 {
				return fmt.Sprintf(col.Type+"(%s)", col.Length.String)
			} else {
				return col.Type
			}
		}
		return col.Type
	default:
		return col.Type
	}
}

// formatDefaultValue 格式化默认值，适配开物数据库语法
func (w *KwdbWriter) formatDefaultValue(value string, dataType string) string {
	// 特殊值处理
	switch strings.ToUpper(value) {
	case "CURRENT_TIMESTAMP", "NOW()":
		return "CURRENT_TIMESTAMP"
	case "NULL":
		return "NULL"
	default:
		// 数值类型不需要引号
		upperType := strings.ToUpper(dataType)
		if strings.Contains(upperType, "INT") ||
			strings.Contains(upperType, "FLOAT") ||
			strings.Contains(upperType, "DOUBLE") ||
			strings.Contains(upperType, "DECIMAL") ||
			strings.Contains(upperType, "NUMERIC") {
			return value
		}

		// 字符串类型需要用引号包围
		if !strings.HasPrefix(value, "'") && !strings.HasPrefix(value, "\"") {
			return fmt.Sprintf("'%s'", strings.ReplaceAll(value, "'", "''"))
		}
		return value
	}
}

// getPrimaryKeyColumns 获取主键列名列表
func (w *KwdbWriter) getPrimaryKeyColumns(columns []common.KwdbColumn) []string {
	var pkColumns []string
	for _, col := range columns {
		if col.PrimaryKey.Valid && col.PrimaryKey.String == "YES" {
			pkColumns = append(pkColumns, `"`+col.Name+`"`)
		}
	}
	return pkColumns
}

// createIndexes 为表创建索引
func (w *KwdbWriter) createIndexes(db *sql.DB, dbName, schemaName, tableName string, indexes []common.IndexInfo) error {
	// 构建完整的表名
	fullTableName := getFullTableName(dbName, schemaName, tableName)

	// 创建索引
	for _, indexInfo := range indexes {
		// 跳过主键索引，因为主键已经在CREATE TABLE语句中定义
		if indexInfo.IsPrimary {
			continue
		}

		// 构建索引名（如果索引名为空，则生成一个）
		indexName := indexInfo.Name
		if indexName == "" {
			// 为索引生成名称
			columnNames := make([]string, len(indexInfo.Columns))
			for i, col := range indexInfo.Columns {
				columnNames[i] = col.Name
			}
			indexName = fmt.Sprintf("idx_%s_%s_%s_%s",
				strings.ReplaceAll(dbName, ".", "_"),
				strings.ReplaceAll(schemaName, ".", "_"),
				strings.ReplaceAll(tableName, ".", "_"),
				strings.Join(columnNames, "_"))
		}

		// 清理索引名中的特殊字符
		indexName = strings.ReplaceAll(indexName, "\"", "")
		indexName = strings.ReplaceAll(indexName, "'", "")

		// 构建列名列表
		columnParts := make([]string, len(indexInfo.Columns))
		for i, col := range indexInfo.Columns {
			columnParts[i] = fmt.Sprintf("\"%s\"", col.Name)
			// 如果指定了排序顺序，则添加
			if col.SortOrder != "" && strings.ToUpper(col.SortOrder) != "ASC" {
				columnParts[i] += " " + strings.ToUpper(col.SortOrder)
			}
		}
		columnsStr := strings.Join(columnParts, ", ")

		// 构建创建索引的SQL语句
		var sql string
		if indexInfo.IsUnique {
			sql = fmt.Sprintf("CREATE UNIQUE INDEX \"%s\" ON %s (%s)", indexName, fullTableName, columnsStr)
		} else {
			sql = fmt.Sprintf("CREATE INDEX \"%s\" ON %s (%s)", indexName, fullTableName, columnsStr)
		}

		// 执行创建索引语句
		_, err := db.Exec(sql)
		if err != nil {
			// 记录日志但不返回错误，因为索引创建失败不应该导致整个表创建失败
			fmt.Printf("Warning: failed to create index %s: %v, SQL: %s\n", indexName, err, sql)
		}
	}
	return nil
}
